package com.ymm.spark.spark.session;

import com.ymm.spark.constant.Constants;
import com.ymm.spark.util.StringUtils;
import org.apache.spark.AccumulatorParam;

/**
 * @author zhangj
 * @date 2019/10/29
 * session聚合统计Accumulator
 */
public class SessionAggrStatAcumulator implements AccumulatorParam<String> {
	/**
	 * 主要是数据初始化
	 * 所有区间范围的数量,都是0
	 * 个个范围数量的拼接,采用各式:key=value|key=value
	 *
	 * @param v
	 * @return
	 */
	@Override
	public String zero(String v) {
		return Constants.SESSION_COUNT + "=0|"
				+ Constants.TIME_PERIOD_1s_3s + "=0|"
				+ Constants.TIME_PERIOD_4s_6s + "=0|"
				+ Constants.TIME_PERIOD_7s_9s + "=0|"
				+ Constants.TIME_PERIOD_10s_30s + "=0|"
				+ Constants.TIME_PERIOD_30s_60s + "=0|"
				+ Constants.TIME_PERIOD_1m_3m + "=0|"
				+ Constants.TIME_PERIOD_3m_10m + "=0|"
				+ Constants.TIME_PERIOD_10m_30m + "=0|"
				+ Constants.TIME_PERIOD_30m + "=0|"
				+ Constants.STEP_PERIOD_1_3 + "=0|"
				+ Constants.STEP_PERIOD_4_6 + "=0|"
				+ Constants.STEP_PERIOD_7_9 + "=0|"
				+ Constants.STEP_PERIOD_10_30 + "=0|"
				+ Constants.STEP_PERIOD_30_60 + "=0|"
				+ Constants.STEP_PERIOD_60 + "=0";
	}

	/**
	 * addAccumulator()和addInPlace()方法可以理解为一样
	 * v1:zero()方法初始化的字符串
	 * v2:遍历每个session的对应区间
	 * 在v1中找到v2对应的value,将value累加1,然后更新回v1
	 *
	 * @param v1
	 * @param v2
	 * @return
	 */
	@Override
	public String addAccumulator(String v1, String v2) {
		return add(v1, v2);
	}


	@Override
	public String addInPlace(String v1, String v2) {
		return add(v1, v2);
	}

	/**
	 * session统计计算逻辑
	 *
	 * @param v1
	 * @param v2
	 * @return
	 */
	public String add(String v1, String v2) {
		if (StringUtils.isEmpty(v1)) {
			return v2;
		}
		//从v1中取出v2对应的值,并累加1
		String oldValue = StringUtils.getFieldFromConcatString(v1, "\\|", v2);
		if (oldValue != null) {
			int newValue = Integer.valueOf(oldValue) + 1;
			//将v1中v2对应的值,设置成新的累加后的值
			return StringUtils.setFieldInConcatString(v1, "\\|", v2, String.valueOf(newValue));
		}
		return v1;
	}

}
